package com.lvmama.rhino.analyze.push


import java.util.{Calendar, Date}

import com.lvmama.rhino.common.entity.{JobConfig, PushMessageReceipt}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions.ImplicitLoad
import com.lvmama.rhino.common.utils.JDBCUtil.JDBCTemplate
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.spark.sql.functions.{col, concat_ws, regexp_replace}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}


/**
  * Created by wuyunpeng on 2016/7/22.
  */
class PushCallbackJob(config: JobConfig, path: String) extends SparkApplication {

  override var appName: String = "PushCallbackJob"
  override var sparkConfig: Map[String, String] = config.spark

  val JDBCDefault = JDBCTemplate.JDBCDefaultSet
  val connP = JDBCTemplate.getConProperties

  val TBL_PUSH_DEVICEINFO: String = "push_deviceInfo"
  val TBL_PUSH_MSG_TRACE: String = "push_msg_trace"
  val TBL_PUSH_MI_REGID: String = "push_mi_regId"
  val FAM_BASEINFO: String = "baseInfo"
  val FAM_BASE_INFO: String = "base_info"


  def execute(): Unit = {

    sparkConfig += ("spark.app.name" -> appName)
    //sparkConfig += ("spark.master" -> "local[2]")
    withSparkContext {
      sc =>
        val sqlContext = new SQLContext(sc)
        import com.lvmama.rhino.common.utils.hbase._
        import sqlContext.implicits._


        val messageRDD = sc.textFile(path).map(_.split("\001")).map(message => PushMessageReceipt(message(0), message(1), message(2), message(3), message(4), message(5), message(6), message(7), message(8), message(9)))
        //ios点击数
        val ios = messageRDD.filter(_.platform == "ios").map(x => (x.tailCode, List((x.getFlag, x.clickFlag))))
          .reduceByKey(_ ++ _).map(x => (x._1, "IPHONE", x._2.count(y => y._1.equals("1")), x._2.count(y => y._2.equals("1"))))
        println(s"ios===${ios.count()}")
        //tailcode platform 到达 点击
        val android = messageRDD.filter(_.platform == "Android").map(x => (x.tailCode, List((x.getFlag, x.clickFlag))))
          .reduceByKey(_ ++ _).map(x => (x._1, "ANDROID", x._2.count(y => y._1.equals("1")), x._2.count(y => y._2.equals("1"))))

        ios.union(android).toDF("tail_code", "platform", "push_deliver_count", "push_click_count")
          .write.mode(SaveMode.Append)
          .option("driver", JDBCDefault("driver"))
          .jdbc(JDBCTemplate.getValue("jdbc.hippo.conn"), "push_message_result", connP)

        //记录明细 至push_msg_trace表中
        //接收消息统计
        val msgDF = messageRDD.map(x => (x.tailCode.replace("pxh", "").reverse, x.tailCode, x.udid, if (x.platform == "ios") "IPHONE" else "ANDROID",
          if (x.messageGetTime.trim.nonEmpty) DateFormatUtils.format(x.messageGetTime.toLong, "yyyy-MM-dd HH:mm:ss") else "", x.getFlag,
          if (x.messageClickTime.trim.nonEmpty) DateFormatUtils.format(x.messageClickTime.toLong, "yyyy-MM-dd HH:mm:ss") else "", x.clickFlag))
          .toDF("msgRowKey", "tailCode", "deviceToken", "platform", "receiveTime", "receiveFlag", "clickTime", "clickFlag")

        //取出7天内的消息数量
        val beginPushTime = DateFormatUtils.format(DateUtils.truncate(DateUtils.addDays(new Date(), -7), Calendar.DATE), "yyyy-MM-dd HH:mm:ss")
        //从mysql中查出消息类型，标题，内容及推送通道
        val tableName = "(select id,begin_time,tail_code,push_category,title,content from push_message_statistic where begin_time >='" + beginPushTime + "') as push_message_statistic "
//        val pushMsgDF: DataFrame = sqlContext.read.option("driver", JDBCDefault("driver")).jdbc(JDBCTemplate.getValue("jdbc.hippo.conn"), tableName, connP).toDF("id", "pushTime", "tailCode", "pushCategory", "title", "content")

        val pushMsgDF: DataFrame = sqlContext
          .loadFromMysqlHippo(tableName)
          .toDF("id", "pushTime", "tailCode", "pushCategory", "title", "content")

        //取出设备及用户信息
        val deviceInfoRdd = sc.hbaseTable[(String, Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
          .select("objectId", "userId")
          .inColumnFamily(FAM_BASEINFO)
          .filter(x => x._2.isDefined && x._3.isDefined)
          .toDF("deviceRowKey", "deviceToken", "userId")

        //生成完整的信息
        val completeMsgDF = msgDF.join(pushMsgDF, Seq("tailCode")).join(deviceInfoRdd, Seq("deviceToken")).
          select(col("id"), col("pushTime"), col("pushCategory"), col("title"), col("content"), col("msgRowKey"), col("tailCode"),
            col("deviceToken"), col("platform"), col("receiveTime"), col("receiveFlag"), col("clickTime"), col("clickFlag"), col("userId"))
          .withColumn("msgTraceRowKey", concat_ws("", col("msgRowKey"), regexp_replace(col("pushTime"), "-", "").substr(0, 8)))

        //收到消息
        completeMsgDF.where(col("receiveFlag") === 1)
          .map(m => (m.get(14).toString, m.get(6).toString, m.get(13).toString, m.get(7).toString, m.get(8).toString,
            m.get(9).toString, m.get(0).toString, m.get(1).toString, m.get(2).toString, m.get(3).toString, m.get(4).toString, "1"))
          .rdd
          .toHBaseTable(TBL_PUSH_MSG_TRACE).toColumns("tailCode", "userId", "deviceToken", "platform", "receiveTime", "pushId", "pushTime", "pushCategory", "title", "content", "receive")
          .inColumnFamily(FAM_BASE_INFO).save()

        //点击的消息
        completeMsgDF.where(col("clickFlag") === 1)
          .map(m => (m.get(14).toString, m.get(6).toString, m.get(13).toString, m.get(7).toString, m.get(8).toString,
            m.get(11).toString, m.get(0).toString, m.get(1).toString, m.get(2).toString, m.get(3).toString, m.get(4).toString, "1"))
          .rdd
          .toHBaseTable(TBL_PUSH_MSG_TRACE).toColumns("tailCode", "userId", "deviceToken", "platform", "clickTime", "pushId", "pushTime", "pushCategory", "title", "content", "click")
          .inColumnFamily(FAM_BASE_INFO).save()

    }
  }
}

object PushCallbackJob {
  val config = JobConfig()

  def apply(path: String): Unit = new PushCallbackJob(config, path).execute()
}